package info.fqtech.weipan;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xiaoleilu.hutool.util.StrUtil;
import info.fqtech.weipan.entity.ForwardDestination;
import info.fqtech.weipan.entity.QuoteHandleRule;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.*;

/**
 * 易享购项目行情的接收和转发
 * Created by admin on 2017/7/10.
 */
public class QuotationReceiveAndForward {

    private static Logger logger = LoggerFactory.getLogger(QuotationReceiveAndForward.class);

    // 转发的地址
    private static List<ForwardDestination> forwardDestinationList = new ArrayList<ForwardDestination>();

    // udp行情监听端口
    private static int port = 35901;

    // 行情发送方ip
    private static String serverIp;

    // 缓冲区大小
    private static int buffSize = 1024;

    // 包含要处理的行情数据
    private static Map<String, QuoteHandleRule> includeQuoteMap = new HashMap<>(8);

    // 存储上一次报价数据
    private static Map<String, String> lastPriceMap = new HashMap<>(8);

    static {
        InputStreamReader in = null;
        try {
            in = new InputStreamReader(QuotationReceiveAndForward.class.getClassLoader().getResourceAsStream("config/forward.properties"), "UTF-8");
            Properties properties = new Properties();
            properties.load(in);
            port = Integer.parseInt(properties.getProperty("port"));
            serverIp = properties.getProperty("serverIp");
            buffSize = Integer.parseInt(properties.getProperty("buff_size"));

            String forwardAddress = properties.getProperty("forwardAddress");
            if (forwardAddress.contains(",")) {
                String[] forwardAddressArray = forwardAddress.split(",");
                for (String str : forwardAddressArray) {
                    fillForwardDestinationList(str);
                }
            }else {
                fillForwardDestinationList(forwardAddress);
            }
            // 行情转发规则
            String handleRules = properties.getProperty("handle_rule");
            fillInIncludeQuoteSet(handleRules);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if(in != null){
                try {
                    in.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 填充转发数据
     * @param forwardStr
     */
    private static void fillForwardDestinationList(String forwardStr){
        String destinationArr[] = forwardStr.split(":");
        String address = destinationArr[0];
        String portStr = destinationArr[1];
        int port = Integer.parseInt(portStr);
        ForwardDestination forwardDestination = new ForwardDestination(address, port);
        forwardDestinationList.add(forwardDestination);
    }

    /**
     * 填充要处理的行情
     * @param handleRules
     */
    private static void fillInIncludeQuoteSet(String handleRules){
        if(handleRules != null && !"".equals(handleRules)){
            if(handleRules.contains(",")){
                String[] handleRuleArray = handleRules.split(",");
                for(String hr : handleRuleArray){
                    addQuoteHandleRule(hr);
                }
            }else {
                addQuoteHandleRule(handleRules);
            }
        }
    }

    /**
     * 添加行情过滤规则
     * @param handleRule
     */
    public static void addQuoteHandleRule(String handleRule){
        String[] array = handleRule.split(":");
        String code = array[0];
        double multiplier = Double.parseDouble(array[1]);
        int digit = Integer.parseInt(array[2]);
        double diff = Double.parseDouble(array[3]);
        QuoteHandleRule quoteHandleRule = new QuoteHandleRule(code, multiplier, digit, diff);
        // 放入要处理的行情集合中
        includeQuoteMap.put(code, quoteHandleRule);
    }


    /**
     * 将需要转发的内容recStr通过udp方式转发出去
     * @param recStr
     */
    private static void forwardData(DatagramSocket datagramSocket, String recStr) {
        try {
            StringBuffer sb = new StringBuffer();
            byte[] buf = (recStr).getBytes("UTF-8"); // 数据
            for (ForwardDestination forwardDestination : forwardDestinationList) {
                InetAddress destination = InetAddress.getByName(forwardDestination.getAddress());
                DatagramPacket dp = new DatagramPacket(buf, buf.length, destination, forwardDestination.getPort());
                if (datagramSocket != null) {
                    // 打包到DatagramPacket类型中（DatagramSocket的send()方法接受此类，注意10000是接受地址的端口，不同于自己的端口！）
                    datagramSocket.send(dp);
                    sb.append(forwardDestination.getAddress() + ":" + forwardDestination.getPort() + "|");
                }
            }
            logger.warn("forward:" + sb.toString() + recStr);
        } catch (SocketException e) {
            logger.error("udp数据转发异常" + e.getMessage());
        } catch (IOException e) {
            logger.error("udp数据转发异常" + e.getMessage());
        }
    }

    /**
     * 获取随机差价
     * @param random
     * @param isPlus
     * @return
     */
    public static double getRandomDiff(Random random, boolean isPlus){
        int i = random.nextInt(2);
        double d = random.nextDouble();
        if(isPlus){
            return (i+d);
        }
        return (i+d) * -1;
    }

    /**
     * 主函数
     * @param args
     */
    public static void main(String[] args) {
        // 随机数产生器
        Random random = new Random();

        DatagramSocket ds = null;
        try {
            ds = new DatagramSocket(port);
            logger.info("udp行监听在端口:" + port + ",服务启动成功");
            boolean stopFlag = false;
            while(!stopFlag){
                byte[] buf = new byte[buffSize];//接受内容的大小，注意不要溢出
                DatagramPacket dp = new DatagramPacket(buf,0,buf.length);//定义一个接收的包
                ds.receive(dp);//将接受内容封装到包中
                String fromServerIp = dp.getAddress().getHostAddress();
                String data = new String(dp.getData(), 0, dp.getLength());//利用getData()方法取出内容
                logger.debug(data.trim());
                if(!serverIp.equals(fromServerIp)){
                    logger.error("非法数据:from:{},data:{}", fromServerIp, data);
                    continue;
                }
                JSONObject jsonObject = JSON.parseObject(data);
                String code = jsonObject.getString("Code");
                String quotePriceStr = jsonObject.getString("Buy");
                if(includeQuoteMap.containsKey(code) && StrUtil.isNotBlank(quotePriceStr)){
                    // 获取处理对象
                    QuoteHandleRule quoteHandleRule = includeQuoteMap.get(code);
                    double quotePrice = Double.parseDouble(quotePriceStr);
                    double newPrice = quotePrice * quoteHandleRule.getMultiplier();
                    if(lastPriceMap.containsKey(code)){
                        // 获取上一个报价
                        String lastPriceStr = lastPriceMap.get(code);
                        double lastPrice = Double.parseDouble(lastPriceStr);
                        // 如果差值>=过滤值
                        double diff = lastPrice - newPrice;
                        if(Math.abs(diff) >= quoteHandleRule.getDiff() || Double.compare(Math.abs(diff), 0.01D) < 0){
                            double randomDiff = 0D;
                            if(diff < 0){
                                // 上次报价比本次报价小，则加正数
                                randomDiff = getRandomDiff(random, true);
                            }else {
                                // 上次报价比本次报价大，则加负数
                                randomDiff = getRandomDiff(random, false);
                            }
                            // 穿插报价，使用上一口报价+随机报价差
                            newPrice = lastPrice + randomDiff;
                            logger.error("last:{}, new:{}, diff:{}, randomDiff:{}", lastPrice, newPrice, diff, randomDiff);
                        }
                    }
                    String newPriceStr = String.format("%." + quoteHandleRule.getDigit() + "f", newPrice);
                    lastPriceMap.put(code, newPriceStr);
                    // 将转换后的报价替换掉原来的报价
                    jsonObject.put("Buy", newPriceStr);
                    // 进行转发
                    forwardData(ds, jsonObject.toJSONString());
                }else {
                    logger.debug("exclude:{}", code);
                }
            }
        } catch (SocketException e) {
            logger.warn("行情接收服务异常退出，" + e.getMessage());
        } catch (IOException e) {
            logger.warn("行情接收服务异常退出，" + e.getMessage());
        } finally {
            if(ds!=null){
                ds.close();
                ds=null;
            }
        }
    }
}
